package com.gandong8.trace.tool.config.kafka;


import com.gandong8.trace.tool.config.GlobalProperties;
import com.gandong8.trace.tool.config.kafka.properties.KafkaProducerProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableConfigurationProperties({KafkaProducerProperties.class, GlobalProperties.class})
@Slf4j
public class KafkaProducerConf {

    @Autowired
    private KafkaProducerProperties producerProperties;

    @Autowired
    private GlobalProperties globalProperties;


    private Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, producerProperties.getServers());
        props.put(ProducerConfig.RETRIES_CONFIG, producerProperties.getRetries());
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, producerProperties.getBatchSize());
        props.put(ProducerConfig.LINGER_MS_CONFIG, producerProperties.getLinger());
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, producerProperties.getBufferMemory());
        props.put(ProducerConfig.CLIENT_ID_CONFIG, producerProperties.getClientId());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, producerProperties.getKeySerializer());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, producerProperties.getValueSerializer());
        props.put(ProducerConfig.ACKS_CONFIG, producerProperties.getAck());
        return props;
    }

    private ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @SuppressWarnings("unchecked")
    @Bean(name="kafkaTemplate", destroyMethod = "destroy")
    public KafkaTemplate<String, String> kafkaTemplate() {
        if(!globalProperties.getKafkaProducerOpen()) {
            System.err.println("kafka producer not allowed");
            return null;
        }
        KafkaTemplate<String, String> kafkaTemplate =  new KafkaTemplate<>(producerFactory());
        //异步处理
        //kafkaTemplate.setProducerListener(new KafkaSendResultHandler());
        log.info("-----kafka producer start!-----");
        return kafkaTemplate;
    }
}
